import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ReactiveVirtualThreads {
    static int count = 500000;
   static CountDownLatch latch = new CountDownLatch(count);

    public static void main(String[] args) throws InterruptedException {
        // 创建虚拟线程执行器
        ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
        long time = System.currentTimeMillis();
        for(int i=0;i<count;++i){
            // 虚拟线程支持的响应式管道
            Flux.range(1, 100)
                    .publishOn(Schedulers.fromExecutor(virtualExecutor))
                    .map(ii -> {
                        // 阻塞操作在虚拟线程中被优化
                        return processItem(ii);
                    })
                    .subscribe(ReactiveVirtualThreads::use);

        }



        System.out.println(System.currentTimeMillis() - time);
    }

    static void use(Object data){

    }

    // 模拟阻塞操作
    static int processItem(int i) {
        try {
            Thread.sleep((long) (Math.random()*5)); // 虚拟线程不会阻塞OS线程
           // latch.countDown();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return i;
    }
}